package com.wan.net.rpc.handler;

import com.wan.net.core.packet.Packet;
import com.wan.net.rpc.message.RpcRequest;
import com.wan.net.rpc.message.RpcResponse;
import com.wan.net.rpc.service.ServiceProvider;
import com.wan.net.rpc.service.ServiceProviderManager;
import com.wan.net.rpc.util.ResultCode;
import com.wan.thread.actor.Actor;
import com.wan.thread.actor.ActorExecutor;
import com.wan.util.serializer.ProtoStuffSerializer;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * @author wwh
 * @date 2022/6/20 18:28
 */
@ChannelHandler.Sharable
public class RpcServerHandler extends SimpleChannelInboundHandler<Packet> {
    private static final Logger logger = LoggerFactory.getLogger(RpcServerHandler.class);
    private static final ActorExecutor actorExecutor;

    static {
        actorExecutor = new ActorExecutor(Runtime.getRuntime().availableProcessors() + 2,"rpc-handler");
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Packet packet) throws Exception {
        Channel channel = channelHandlerContext.channel();
        byte[] content = packet.getContent();
        RpcRequest rpcRequest = ProtoStuffSerializer.decode(content, RpcRequest.class);
        try {
            ServiceProvider provider = ServiceProviderManager.getProvider(rpcRequest.getServiceName());
            if(provider == null){
                logger.error("[rpc]找不到服务方法:{}",rpcRequest.getServiceName());
                response(channel,packet,ResultCode.SERVICE_NOT_EXIST);
                return;
            }
            final long theadId = rpcRequest.getTempId() == 0 ? actorExecutor.autoId() : rpcRequest.getTempId();
            actorExecutor.execute(new Actor() {
                @Override
                public long asyncId() {
                    return theadId;
                }

                @Override
                public void run() {
                    try {
                        Object result = provider.invoke(rpcRequest.getParams());
                        response(channel,packet,ResultCode.SUCCESS,result);
                    }catch (Exception e){
                        logger.error("[rpc]服务调用时出错,targetService:{},params:{}",rpcRequest.getServiceName(),rpcRequest.getParams(),e);
                        try {
                            RpcServerHandler.response(channel,packet,ResultCode.INVOKE_ERROR);
                        } catch (IOException ioException) {
                            logger.error("[rpc]响应失败",ioException);
                        }
                    }
                }
            });
        }catch (Exception e){
            logger.error("[rpc]消息处理出错,targetService:{},params:{}",rpcRequest.getServiceName(),rpcRequest.getParams(),e);
            response(channel,packet,ResultCode.INVOKE_ERROR);
        }
    }

    /**
     * 回复消息
     * @param resultCode 错误码
     */
    private static void response(Channel channel,Packet packet,ResultCode resultCode,Object result) throws IOException {
        RpcResponse response = new RpcResponse();
        response.setCode(resultCode.getCode());
        response.setResult(result);
        byte[] encode = ProtoStuffSerializer.encode(response);
        packet.setContent(encode);
        channel.writeAndFlush(packet);
    }

    private static void response(Channel channel,Packet packet,ResultCode resultCode) throws IOException {
        response(channel,packet,resultCode,null);
    }

    /**
     * 事件触发
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            logger.error("释放空闲的[rpc]连接...channel={}",ctx.channel());
            ctx.close();
        }
    }

    /**
     * 异常
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error("连接异常...channel={}", ctx.channel(),cause);
        ctx.close();
    }
}
